/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.common.ipc;

import java.io.IOException;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.nhncorp.neptune.common.Constants;
import com.nhncorp.neptune.common.LockServiceFactory;
import com.nhncorp.neptune.common.conf.NConfiguration;
import com.nhncorp.neptune.common.exception.PermissionException;
import com.nhncorp.neptune.tablet.TableSchema;
import com.nhncorp.pleiades.client.LockService;
import com.nhncorp.pleiades.protocol.LockMetaData;

public class AclManager {
  private static final Log LOG = LogFactory.getLog(AclManager.class.getName());

  public static ThreadLocal<String> current = new ThreadLocal<String>();

  private static String superGroupUser;
  
  private static Object monitor = new Object();
  
  //private static NConfiguration conf = new NConfiguration();
  
  public static boolean isSuperGroup(LockService lockService) throws IOException {
    String userId = current.get();
    return isSuperGroup(lockService, userId);
  }
  
  public static boolean isSuperGroup(LockService lockService, String userId) throws IOException {
    synchronized(monitor) {
      if(superGroupUser == null) {
        LockMetaData userLockMetaData = lockService.getMetaData(Constants.SUPERGROUP);
        if(userLockMetaData == null) {
          return false;
        } else {
          superGroupUser = userLockMetaData.getContentKey();
        }
      }
    }
    if (userId == null || userId.trim().length() == 0 || !userId.equals(superGroupUser)) {
      return false;
    }

    return true;
  }

  public static void checkOwner(NConfiguration conf, LockService lockService,
      Map<String, TableSchema> tables, String tableName)
      throws IOException {
    String userId = current.get();

    if (userId == null || userId.trim().length() == 0) {
      throw new PermissionException("No permission(userId is null)");
    }

    if (isSuperGroup(lockService, userId)) {
      return;
    }

    synchronized (tables) {
      if (!tables.containsKey(tableName)) {
        TableSchema tableSchema = TableSchema.loadTableSchema(conf, lockService, tableName);
        if (tables != null) {
          tables.put(tableName, tableSchema);
        } else {
          throw new PermissionException("No permission[user=" + userId
              + ", table=" + tableName + "]");
        }
      }
    }

    TableSchema tableSchema = null;
    synchronized (tables) {
      tableSchema = tables.get(tableName);
    }

    if (userId == null || userId.trim().length() == 0) {
      throw new PermissionException("No permission[user=" + userId + ", table="
          + tableName + "]");
    }
    
    if (!tableSchema.getOwner().equals(userId)) {
      throw new PermissionException("No permission[user=" + userId + ", table="
          + tableName + "]");
    }
  }

  public static void checkPermission(NConfiguration conf, LockService lockService,
      Map<String, TableSchema> tables, String tableName, String readWrite)
      throws IOException {
    String userId = current.get();
    checkPermission(conf, lockService, userId, tables, tableName, readWrite);
  }

  public static void checkPermission(NConfiguration conf, LockService lockService, 
      String userId, Map<String, TableSchema> tables, 
      String tableName, String readWrite) throws IOException {
    if (userId == null || userId.trim().length() == 0) {
      throw new PermissionException("No permission(userId is null)");
    }

    if (isSuperGroup(lockService, userId)) {
      return;
    }

    synchronized (tables) {
      if (!tables.containsKey(tableName)) {
        TableSchema tableSchema = TableSchema.loadTableSchema(conf, lockService, tableName);
        if (tables != null) {
          tables.put(tableName, tableSchema);
        } else {
          throw new PermissionException("No permission[user=" + userId
              + ", table=" + tableName + ", op=" + readWrite + "]");
        }
      }
    }

    TableSchema tableSchema = null;
    synchronized (tables) {
      tableSchema = tables.get(tableName);
    }

    if (userId == null || userId.trim().length() == 0) {
      throw new PermissionException("No permission[user=" + userId + ", table="
          + tableName + ", op=" + readWrite + "]");
    }
    if (!tableSchema.hasPermission(userId, readWrite)) {
      throw new PermissionException("No permission[user=" + userId + ", table="
          + tableName + ", op=" + readWrite + "]");
    }
  }

  public static void checkUser(LockService lockService,
      Set<String> users, String paramUserId)
      throws IOException, PermissionException {
    if (users == null) {
      users = new HashSet<String>();
    }
    synchronized (users) {
      if(users.contains(paramUserId)) {
        return;
      }
      LockMetaData userMeta = lockService.getMetaData(Constants.USERS + "/"
          + paramUserId);
      if (userMeta == null) {
        throw new PermissionException("No user info:" + paramUserId);
      }
      users.add(paramUserId);
      return;
    }
  }
  
  public static Set<String> loadUserInfo(NConfiguration conf, LockService lockService) throws IOException {
    Set<String> users = new HashSet<String>();
    Map<String, LockMetaData> metas = lockService.readDir(Constants.USERS);
    if(metas != null && metas.size() > 0) {
      for(LockMetaData eachMeta: metas.values()) {
        String userId = LockServiceFactory.getLockMetaDataPathName(eachMeta);
        users.add(userId);
      }
    }
    
    return users;
  }

//  public static String getRootPasswd(NeptuneConfiguration conf) throws IOException {
//    GPath path = new GPath(conf.get("neptune.root") + "/" + Constants.USERS + "/root");
//    NeptuneFileSystem fs = NeptuneFileSystem.get(conf);
//    if(!fs.exists(path)) {
//      LOG.fatal("No root user info:" + path);
//      return null;
//    }
//    
//    try {
//      Thread.sleep(2 * 1000);
//    } catch (InterruptedException e1) {
//      return null;
//    }
//    BufferedReader reader = null;
//    try {
//      reader = new BufferedReader(new InputStreamReader(fs.open(path)));
//      String line = reader.readLine();
//      if(line == null) {
//        LOG.fatal("No root user info: data is null:" + path);
//        return null;
//      }
//      LOG.fatal("getRootPasswd:" + line);
//      return CipherUtil.decrypt(conf, line);
//    } catch (Exception e) {
//      LOG.error(e);
//      return null;
//    } finally {
//      reader.close();
//    }
//    
//  }
//  
//  public static void saveRootPasswd(NeptuneConfiguration conf, String keyPassword, String passwd) throws IOException {
//    GPath path = new GPath(conf.get("neptune.root") + "/" + Constants.USERS + "/root");
//    NeptuneFileSystem fs = NeptuneFileSystem.get(conf);
//    if(fs.exists(path)) {
//      fs.delete(path);
//    }
//    
//    OutputStream out = fs.create(path);
//    String encodedPasswd = CipherUtil.encrypt(conf, keyPassword, passwd);
//    out.write(encodedPasswd.getBytes());
//    out.close();
//    
//    fs.setPermission(path, (short)700);
//  }
  
//  public static Map<String, String> loadUserInfo(NeptuneConfiguration conf) throws IOException {
//    Map<String, byte[]> result = new HashMap<String, byte[]>();
//    NeptuneFileSystem fs = NeptuneFileSystem.get(conf);
//    GPath path = getUserListFilePath(conf);
//    if(!fs.exists(path)) {
//      return result; 
//    }
//    
//    DataInputStream in = null;
//    try {
//      in = new DataInputStream(fs.open(path));
//      while(true) {
//        try {
//          String userId = WritableUtils.readString(in);
//          int length = in.readInt();
//          byte[] passwd = new byte[length];
//          in.readFully(passwd, 0, length);
//          result.put(userId, passwd);
//        } catch (EOFException e) {
//          break;
//        }
//      }
//    } finally {
//      if(in != null) {
//        in.close();
//      }
//    }
//    return result;
//  }
//  
//  public static void saveUsersInfo(NeptuneConfiguration conf, Map<String, byte[]> users) throws IOException {
//    NeptuneFileSystem fs = NeptuneFileSystem.get(conf);
//    GPath path = getUserListFilePath(conf);
//    fs.delete(path);
//    DataOutputStream out = null;
//    try {
//      out = new DataOutputStream(fs.create(path));
//      for(Map.Entry<String, byte[]> entry: users.entrySet()) {
//        WritableUtils.writeString(out, entry.getKey());
//        byte[] passwd = entry.getValue();
//        out.writeInt(passwd.length);
//      }
//    } finally {
//      if(out != null) {
//        out.close();
//      }
//    }
//  }
//  
//  public static GPath getUserListFilePath(NeptuneConfiguration conf) {
//    return new GPath(conf.get("neptune.root") + Constants.USERS);
//  }  
}
